package cn.seecoder.ai.service

import cn.seecoder.ai.dao.FileFieldRepository
import cn.seecoder.ai.model.vo.{FileFieldVO, FileInfoVO}
import cn.seecoder.ai.service.{FileFieldService, FileInfoService}
import cn.seecoder.ai.utils.HdfsHelper
import lombok.extern.slf4j.Slf4j
import org.apache.spark.sql.functions.{col, count, desc}
import org.apache.spark.sql.types.DataType
import org.apache.spark.sql.{DataFrame, SaveMode, SparkSession}
import org.slf4j.{Logger, LoggerFactory}
import org.springframework.beans.factory.annotation.Autowired
import org.springframework.stereotype.Service
import cn.seecoder.ai.model.vo.FileInfoVO

import java.util



/**
 * 对文件字段进行分析的工具类，逻辑为：在确保文件为parquet的情况下，对每个字段进行分析。
 * todo: 添加众数的获取，考虑后期填充使用众数填充。
 * @author   fanyanpeng
 * @date 2023/4/20 4:21
 * @param null
 * @return
 */
@Service
class FileFieldService {
  private val log: Logger = LoggerFactory.getLogger(getClass)

  @Autowired
  private var fileInfoService: FileInfoService = _

  @Autowired
  private var sparkSession: SparkSession = _

  @Autowired
  private var hdfsHelper: HdfsHelper = _

  @Autowired
  private var fileFieldRepository: FileFieldRepository = _

  /**
   * 确保文件类型已经完成转化，两条路径：
   * 1.若fileUri为空，就需要进行转换：
   * 读取originalFileUri，生成parquet文件保存位置，使用spark读取csv文件并修改格式保存。
   * 2.若fileUri不为空，直接返回
   *
   * @author fanyanpeng
   * @date 2023/4/12 22:38
   * @param fileInfoVO
   * @return void
   */
  private def ensureFileConverted(fileInfoVO: FileInfoVO): Boolean = {
    log.info("确保文件已经转换格式：" + fileInfoVO.toString)


    log.info("进入转换流程")

    val userId = fileInfoVO.getUserId
    val originalFileUri: String = fileInfoVO.getOriginalFileUri

    //读取文件到DataFrame
    val df = sparkSession.read
      .option("header", "true")
      .option("inferSchema", "true")
      .csv(originalFileUri)
      .persist()
    log.info("文件csv文件读取成功")
    // 显示前十行
    df.show(10)

    val fileName = fileInfoVO.getFileName;
    //通过hdfs助手获取转换未parquet文件应该存放的位置
    val fileUri: String = hdfsHelper.getUriConvertToParquetFile(userId, fileName)

    //使用spark将文件以parquet格式保存，（使用parquet文件作为系统通用文件处理格式，列存储，加载存储效率高）
    df.coalesce(1).write.mode(SaveMode.Overwrite).parquet(fileUri)

    log.info("文件保存为parquet格式成功")

    // 将存储结果写回数据库
    fileInfoVO.setFileUri(fileUri);
    fileInfoService.saveFileInfoVO(fileInfoVO)

  }

  /**
   * 分析文件字段, 必须进行分析才能进入下一阶段！
   *
   * @author fanyanpeng
   * @date 2023/4/12 16:03
   * @param fileInfoVO
   * @return boolean
   */
  def analysisField(userId: Integer, fileInfoId: Integer): Boolean = {
    // 取出现有文件信息
    val fileInfoVO: FileInfoVO = fileInfoService.getFileInfo(userId, fileInfoId)
    // 确保文件格式已转换
    ensureFileConverted(fileInfoVO)
    //经过上面的操作，已经确保fileUri被填写
    val fileUri = fileInfoVO.getFileUri
    // 将远端parquet文件读取到内存
    val df: DataFrame = sparkSession.read
      .parquet(fileUri)
      .persist()

    // Print the schema in a tree format
    df.printSchema()
    df.show(10)

    //对每一列，分别处理：
    // 1.获取列名称
    // 2.获取类型
    // 3.获取空值率
    log.info("打印")
    val fileFieldVOList: util.List[FileFieldVO] = new util.ArrayList[FileFieldVO]()

    for (column: String <- df.columns) {
      val structField = df.schema(column)
      val dataType = structField.dataType
      val fieldName = structField.name

      //计算null比率
      val nullRowCount = df.filter(df(column).isNull).count() // 使用 xxx is null，若为中文，就会出错，但是目前的写法不会
      val totalRowCount = df.count()
      val nullRatio = nullRowCount * 100.0 / totalRowCount

      //众数数据集，正常情况包括一行一列
      val modeDataSet = df.filter(df(column).isNotNull)
        .groupBy(col(column))
        .agg(count("*").alias("count"))
        .orderBy(desc("count"))
        .limit(1)
        .select(column)
        .collect()

      // 若没有数据，传入null
      var mode: Any = null
      if (modeDataSet.length == 0) {
        mode = null
      }
      else {
        mode = modeDataSet.apply(0).apply(0)
      }

      println(s"Mode is: $mode")
      println(s"Column Name: $column")
      println(s"Data Type: $dataType")
      println(f"Null Percentage: $nullRatio%.2f%%")

      fileFieldVOList.add(new FileFieldVO()
        .setFieldName(fieldName)
        .setFileInfoId(fileInfoId)
        .setDataType(dataType.typeName)
        .setNullRate(nullRatio)
        .setMode(mode)
      )
    }
    log.info("字段分析完成，开始保存")
    fileInfoVO.setFileFields(fileFieldVOList)
    //先清空当前文件下的字段分析结果，再保存
    fileFieldRepository.deleteByFileInfoId(fileInfoId)
    fileInfoService.saveFileInfoVO(fileInfoVO)
  }

}
